#include "vzconn/conn/vzconnect.h"
#include "vzconn/pool/perfectbufferpool.h"

namespace vzconn {

VzConnect::VzConnect(boost::asio::io_service &io_service)
  : io_service_(io_service),
    state_(CS_CLOSED),
    is_writing_(false),
    read_buffer_(new ByteBuffer(ByteBuffer::ORDER_HOST)) {
}

VzConnect::~VzConnect() {
}

void VzConnect::HandleReadData(const boost::system::error_code& err,
                               int read_size) {

  if (state_ == CS_CLOSED) {
    return;
  }

  if (err) {
    LOG(ERROR) << err.message();
    return ConnectError(err);
  }

  read_buffer_->WriteBytes(pre_read_buffer_, read_size);

  // -------------------------------------------------------------------------
  while (read_buffer_->Length() >= HEADER_BUFFERR_SIZE) {
    memcpy(read_header_buffer_, read_buffer_->Data(), HEADER_BUFFERR_SIZE);
    if (read_header_buffer_[0] != VZ_HEADER[0]
        || read_header_buffer_[1] != VZ_HEADER[1]) {
      LOG(ERROR) << "The packet header VZ getting error";
      return ConnectError(boost::asio::error::invalid_argument);
    }
    uint16 flag = *((uint16*)(read_header_buffer_ + 2));
    uint32 packet_size = *(uint32*)(read_header_buffer_ + 4);
    flag = ntohs(flag);
    packet_size = ntohl(packet_size);
    if (packet_size > MAX_PKT_SIZE) {
      LOG(ERROR) << "The packet size large than " << MAX_PKT_SIZE;
      return ConnectError(boost::asio::error::invalid_argument);
    }
    if (packet_size <= (read_buffer_->Length() - HEADER_BUFFERR_SIZE)) {
      SignalConnectRead(shared_from_this(),
                        read_buffer_->Data() + HEADER_BUFFERR_SIZE,
                        packet_size, flag);
      read_buffer_->Consume(packet_size + HEADER_BUFFERR_SIZE);
    } else {
      break;
    }
  }
  AsyncReadData();
}

void VzConnect::HandleAsyncWrite(ByteBuffer::Ptr buffer, uint16 flag) {
  if (state_ != CS_CONNECTED) {
    LOG(ERROR) << "The socket is not opened";
    return;
  }
  buffers_.push_back(buffer);
  TryToWriteData();
}

bool VzConnect::AsyncWrite(const char* data, int size, int flag) {
  // LOG(INFO) << "state_ = " << state_;
  if (state_ != CS_CONNECTED) {
    LOG(ERROR) << "The socket is not opened";
    return false;
  }

  if (size > MAX_PKT_SIZE) {
    LOG(ERROR) << "The size of data is error";
    return false;
  }

  ByteBuffer::Ptr buffer = PerfectBufferPool::Instance()->TakeBuffer(size);
  BOOST_ASSERT(buffer->size() == 0);
  buffer->WriteBytes(VZ_HEADER, 2);
  buffer->WriteUInt16(flag);
  buffer->WriteUInt32(size);
  if(size != 0) {
    buffer->WriteBytes(data, size);
  } else {
    LOG(WARNING) << "Only send header, the data size is zero";
  }
  io_service_.post(
    boost::bind(&VzConnect::HandleAsyncWrite,
                shared_from_this(),
                buffer, flag));
  return true;
}

bool VzConnect::AsyncWrite(ByteBuffer::Ptr bytebuffer, int flag) {

  if (state_ != CS_CONNECTED) {
    LOG(ERROR) << "The socket is not opened";
    return false;
  }

  if (bytebuffer->size() > MAX_PKT_SIZE) {
    return false;
  }

  ByteBuffer::Ptr buffer = PerfectBufferPool::Instance()->TakeBuffer(bytebuffer->size());
  BOOST_ASSERT(buffer->size() == 0);
  buffer->WriteBytes(VZ_HEADER, 2);
  buffer->WriteUInt16(flag);
  buffer->WriteUInt32(bytebuffer->size());
  buffer->WriteBytes(bytebuffer->Data(), bytebuffer->size());
  io_service_.post(
    boost::bind(&VzConnect::HandleAsyncWrite,
                shared_from_this(),
                buffer, flag));
  return true;
}

void VzConnect::HandleDataWrite(const boost::system::error_code& err,
                                ByteBuffer::Ptr write_buffer,
                                int write_size) {

  if (state_ == CS_CLOSED) {
    return;
  }

  if (err) {
    LOG(ERROR) << err.message();
    return ConnectError(err);
  }
  is_writing_ = false;

  if (!write_buffer->Consume(write_size)) {
    LOG(ERROR) << "Never reach here";
    return ConnectError(boost::asio::error::invalid_argument);
  }
  if (write_buffer->size() > 0) {
    // Continue write data
    TryToWriteData();
  } else if (write_buffer->size() == 0) {
    // LOG(INFO) << write_size;
    BOOST_ASSERT(buffers_.size() > 0);
    PerfectBufferPool::Instance()->RecyleBuffer(write_buffer);
    buffers_.erase(buffers_.begin());
    if(buffers_.size() == 0) {
      SignalConnectWrite(shared_from_this());
    } else {
      TryToWriteData();
    }
  } else {
    BOOST_ASSERT(0);
  }
}

void VzConnect::CloseConnect() {

  if (state_ == CS_CLOSED) {
    DLOG(WARNING) << "The state is CS_CLOSED";
    return;
  }
  DLOG(INFO) << "Close connect";
  state_ = CS_CLOSED;
  io_service_.post(
    boost::bind(&VzConnect::HandleCloseConnect, shared_from_this()));
}


////////////////////////////////////////////////////////////////////////////////
BlockVzConnect::BlockVzConnect(boost::asio::io_service &io_service)
  : io_service_(io_service) {
}

BlockVzConnect::~BlockVzConnect() {
}

bool BlockVzConnect::SyncWrite(const char *data, uint32 size, uint16 flag) {

  header_buffer_[0] = 'V';
  header_buffer_[1] = 'Z';
  flag = htons(flag);
  uint32 nsize = htonl(size);
  memcpy(header_buffer_ + 2, &flag, sizeof(uint16));
  memcpy(header_buffer_ + 4, &nsize, sizeof(uint32));

  std::vector<boost::asio::const_buffer> buffers;
  buffers.push_back(boost::asio::buffer(header_buffer_, HEADER_BUFFERR_SIZE));
  buffers.push_back(boost::asio::buffer(data, size));
  return RealySyncWrite(buffers);
}

bool BlockVzConnect::SyncRead(std::vector<char> &read_buffer, uint16 *flag) {
  // 1. Read the header
  if(!RealySyncRead(header_buffer_, HEADER_BUFFERR_SIZE)) {
    LOG(ERROR) << "Read header error";
    return false;
  }
  // 2. analisys the header
  uint16 packet_flag = 0;
  memcpy(&packet_flag, header_buffer_ + 2, sizeof(uint16));
  packet_flag = ntohs(packet_flag);
  if(flag) {
    *flag = packet_flag;
  }
  // -
  uint32 packet_size = 0;
  memcpy(&packet_size, header_buffer_ + 4, sizeof(uint32));
  packet_size = ntohl(packet_size);
  if(packet_size == 0) {
    return true;
  }
  read_buffer.resize(packet_size);
  // 3. read the packet data
  return RealySyncRead(&read_buffer[0], packet_size);
}

}
